#!/bin/bash

TOPIC1='ceph'
TOPIC2='block'
TOPIC3='QuintetQueue'
IP="192.168.1.101"
port="9092"
zookeeperPort="2181"
topics=()
kafkaName="kafka"
zookeeperName="zookeeper"

usage="usage: $0 -i/--IP <IP> -p/--port <port> -zp/--zookeeperPort <zookeeperPort> -t/--topic <topic> ...\n"
usage=$usage"\t-i, --IP: kafka broker's IP\n"
usage=$usage"\t-p, --port: kafka broker's port\n"
usage=$usage"\t-zp, --zookeeperPort: zookeeper's port\n"
usage=$usage"\t-t, --topic: kafka topic\n"

usage_exit() {
    printf "$usage"
    exit
}

# if [ $# -gt 3 ]; then
#     usage_exit
# fi

while [ $# -ge 1 ]; do
case $1 in
    -i | --IP)
        [ -z "$2" ] && usage_exit
        IP=$2
        shift
        ;;
    -p | --port)
        [ -z "$2" ] && usage_exit
        port=$2
        shift
        ;;
    -zp | --zookeeperPort)
        [ -z "$2" ] && usage_exit
        zookeeperPort=$2
        shift
        ;;
    -t | --topic)
        [ -z "$2" ] && usage_exit
        # add to topics array
        topics[${#topics[*]}]=$2
        shift
        ;;
    *)
        usage_exit
esac
shift
done

kafkaName=$kafkaName"_"$port
zookeeperName=$zookeeperName"_"$zookeeperPort

echo "kafka IP: $IP, port: $port, zookeeperPort: $zookeeperPort"
echo "kafkaName: $kafkaName, zookeeperName: $zookeeperName"
echo "topics: $TOPIC1 $TOPIC2 $TOPIC3 ${topics[*]}"

# set -x
echo "stop and remove kafka and zookeeper container"
zookeeper_ids=`docker ps -aq --filter name="${zookeeperName}"`
if [ 0"$zookeeper_ids" != "0" ]; then
    docker stop $(docker ps -aq --filter name="${zookeeperName}")
    docker rm $(docker ps -aq --filter name="${zookeeperName}")
fi

kafka_ids=`docker ps -aq --filter name="${kafkaName}"`
if [ 0"$kafka_ids" != "0" ]; then
    docker stop $(docker ps -aq --filter name="${kafkaName}")
    docker rm $(docker ps -aq --filter name="${kafkaName}")
fi

echo "start kafka and zookeeper container"
docker run -d --name ${zookeeperName} --publish ${zookeeperPort}:2181 \
    --volume /etc/localtime:/etc/localtime wurstmeister/zookeeper
docker run -d --name ${kafkaName} --publish ${port}:${port} --link ${zookeeperName} \
    --env KAFKA_ZOOKEEPER_CONNECT=${IP}:${zookeeperPort} \
    --env KAFKA_ADVERTISED_HOST_NAME=${IP} \
    --env KAFKA_ADVERTISED_PORT=${port} \
    --env KAFKA_PORT=${port} \
    --volume /etc/localtime:/etc/localtime wurstmeister/kafka

sleep 10s

kafka_version=`docker exec ${kafkaName} /bin/bash -c "ls /opt | grep kafka_"`

# create a topic: ceph, saving ceph obj hash which will put fabric
docker exec ${kafkaName} /opt/${kafka_version}/bin/kafka-topics.sh --create --zookeeper ${IP}:${zookeeperPort} --replication-factor 1 --partitions 1 --topic $TOPIC1
# create a topic: block, containning obj ids in a block which using by redundancy module
docker exec ${kafkaName} /opt/${kafka_version}/bin/kafka-topics.sh --create --zookeeper ${IP}:${zookeeperPort} --replication-factor 1 --partitions 1 --topic $TOPIC2
# create a topic: QuintetQueue
docker exec ${kafkaName} /opt/${kafka_version}/bin/kafka-topics.sh --create --zookeeper ${IP}:${zookeeperPort} --replication-factor 1 --partitions 1 --topic $TOPIC3
# create topics from topics array
for topic in ${topics[*]}; do
    docker exec ${kafkaName} /opt/${kafka_version}/bin/kafka-topics.sh --create --zookeeper ${IP}:${zookeeperPort} --replication-factor 1 --partitions 1 --topic $topic
done

echo "created topics: $TOPIC1 $TOPIC2 $TOPIC3 ${topics[*]}"
# list topics
docker exec ${kafkaName} /opt/${kafka_version}/bin/kafka-topics.sh --list --zookeeper ${IP}:${zookeeperPort}

# set data retain time
# docker exec ${kafkaName} /opt/${kafka_version}/bin/kafka-configs.sh --bootstrap-server localhost:${port} \
# --alter --entity-default --entity-type brokers --add-config 'log.retention.ms=3600000'
# docker exec ${kafkaName} /opt/${kafka_version}/bin/kafka-configs.sh --zookeeper ${IP}:${zookeeperPort} \
# --alter --entity-name mytopic --entity-type topics --add-config retention.ms=3600000

# cat topic setting
# docker exec ${kafkaName} /opt/${kafka_version}/bin/kafka-configs.sh --bootstrap-server localhost:${port} --describe --entity-default --entity-type brokers
# docker exec ${kafkaName} /opt/${kafka_version}/bin/kafka-configs.sh --zookeeper ${IP}:${zookeeperPort} --describe --entity-name mytopic --entity-type topics

# run a producer
# docker exec -it ${kafkaName} /opt/${kafka_version}/bin/kafka-console-producer.sh --broker-list localhost:${port} --topic $TOPIC

# run a consumer
# docker exec ${kafkaName} /opt/${kafka_version}/bin/kafka-console-consumer.sh --bootstrap-server localhost:${port} --topic $TOPIC

# docker exec ${kafkaName} bash /opt/kafka_2.13-2.8.1/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:${port} --topic ceph --time -1